# coning utf-8

from kafka import KafkaConsumer
import pandas as pd
from pymongo import MongoClient


def read_data():
    list_data = []

    consumer = KafkaConsumer(
        'JD',
        bootstrap_servers='hadoop101:9092',
        auto_offset_reset='earliest',
        consumer_timeout_ms=1000,  # 设置超时时间为1000毫秒
    )

    for message in consumer:
        utf8_string: str = message.value.decode('utf-8')
        data = utf8_string.split(',')
        list_data.append(data)

    consumer.close()  # 关闭消费者连接
    return list_data


def wash(list_data):
    df = pd.DataFrame(list_data, columns=['品牌', '价格', '售出', '产品', '标题'])

    df.drop(columns=['标题'], inplace=True)  # 排除标题

    # 数据清洗
    df['价格'] = df['价格'].apply(lambda x: str(x).replace('￥', '').replace('￥ ', '').strip()[:-1])
    df['价格'] = pd.to_numeric(df['价格'])

    df['售出'] = df['售出'].apply(
        lambda x: int(x.replace('累计评价', '').replace('+', '').replace('万', '0000').strip()))

    df['产品'] = df['产品'].apply(lambda x: x.replace('商品名称：', ''))

    print(df.shape)
    df.drop_duplicates(inplace=True)  # 去重
    print(df.head())
    print(df.shape)
    return df


def to_Mongo(df):
    # 连接到MongoDB
    client = MongoClient('mongodb://localhost:27017/')
    db = client['JD']  # 替换为你的数据库名
    collection = db['JD_data']  # 替换为你的集合名

    # 将DataFrame转换为字典列表
    records = df.to_dict('records')

    # 插入数据到MongoDB
    collection.insert_many(records)

    print("数据已成功存储到MongoDB中。")
    client.close()  # 关闭数据库连接


if __name__ == '__main__':
    list_data = read_data()
    df = wash(list_data)
    to_Mongo(df)
